package com.geniusk.wss;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.geniusk.common.enums.ResponseType;
import com.geniusk.common.exception.GraceException;
import com.geniusk.common.utils.RedisUtil;
import com.geniusk.common.utils.SpringContextUtil;
import com.geniusk.mqtt.utils.MqttUtil;
import com.geniusk.security.core.LoginUser;
import com.geniusk.security.utils.TokenUtil;
import com.geniusk.system.model.pojo.Device;
import com.geniusk.system.model.pojo.ModelProperty;
import com.geniusk.system.service.DeviceService;
import com.geniusk.system.service.ModelPropertyService;
import com.geniusk.wss.core.BrokerConstants;
import com.geniusk.wss.core.DataContent;
import com.geniusk.wss.core.TopicConstants;
import com.geniusk.wss.enums.MsgActionEnum;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.json.JSONArray;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * 处理消息的handler
 * TextWebSocketFrame： 在netty中，是用于为websocket专门处理文本的对象，frame是消息的载体
 */
@Slf4j
@Component
public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

	private final ModelPropertyService modelPropertyService = (ModelPropertyService) SpringContextUtil.getBean("modelPropertyServiceImpl");

	private final DeviceService deviceService = (DeviceService) SpringContextUtil.getBean("deviceServiceImpl");

	// 用于记录和管理所有客户端的channel
	public static final ChannelGroup wssChannelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

	private static final TokenUtil tokenUtil = SpringContextUtil.getBean(TokenUtil.class);

	private final RedisUtil redisUtil = SpringContextUtil.getBean(RedisUtil.class);

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
		// 获取客户端传输过来的消息
		String content = msg.text();
		Channel currentChannel = ctx.channel();
		// 获取客户端发来的消息
		DataContent dataContent = JSONUtil.parse(content).toBean(DataContent.class);
		Integer action = dataContent.getAction();
		// 判断消息类型，根据不同的类型来处理不同的业务
		if (action.equals(MsgActionEnum.CONNECT.type)) {
			// 当websocket 第一次open的时候，初始化channel，把用的channel和userID关联起来
			String token = dataContent.getToken();
			Long identityId = getIdentityIdByToken(token);
			UserChannelSession.setChannel(identityId, currentChannel);
			Long deviceId = dataContent.getDeviceId();
			//订阅数据关联
			Set<Long> identityIdList = redisUtil.getCacheObject(BrokerConstants.DEVICE_SUB_IDENTITY + deviceId);
			Set<Long> deviceIdList = redisUtil.getCacheObject(BrokerConstants.IDENTITY_SUB_DEVICE + identityId);
			if(identityIdList == null){
				identityIdList = new HashSet<>();
			}
			identityIdList.add(identityId);
			redisUtil.setCacheObject(BrokerConstants.DEVICE_SUB_IDENTITY + deviceId, identityIdList);
			if(deviceIdList == null){
				deviceIdList = new HashSet<>();
			}
			deviceIdList.add(deviceId);
			redisUtil.setCacheObject(BrokerConstants.IDENTITY_SUB_DEVICE + identityId, deviceIdList);
			UserChannelSession.output();
		} else if (action.equals(MsgActionEnum.MESSAGE.type)) {
			String token = dataContent.getToken();
			Long identityId = getIdentityIdByToken(token);
			// 发送消息到设备
			Long deviceId = dataContent.getDeviceId();
			String message = dataContent.getMessage();
			Device device = deviceService.selectDeviceById(deviceId);
			Long productId = device.getModelId();
			// 平台下发行为主题
			String topicName = productId + "/" + deviceId + "/"+TopicConstants.ACTION_TOPIC;
			MqttUtil.sendToMqtt(topicName, message);
		} else if (action.equals(MsgActionEnum.KEEPALIVE.type)) {
			currentChannel.id().asLongText();
		} else {
			log.info("无效连接 {}", currentChannel.id().asLongText());
		}

	}

	/**
	 * 当客户端连接服务端之后（打开连接）
	 * 获取客户端的channel，并且放到ChannelGroup中去进行管理
	 */
	@Override
	public void handlerAdded(ChannelHandlerContext ctx) {
		String channelId = ctx.channel().id().asLongText();
//		log.info("客户端已连接，channelId为：{}", channelId);
		wssChannelGroup.add(ctx.channel());
	}

	@Override
	public void handlerRemoved(ChannelHandlerContext ctx) {
		String channelId = ctx.channel().id().asLongText();
//		log.info("客户端被移除，channelId为：{}", channelId);
		// 当触发handlerRemoved，ChannelGroup会自动移除对应客户端的channel
		removeIdentityId(channelId);
	}

	/**
	 * 捕获异常
	 */
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		log.error(cause.getMessage());
		Channel receiverChannel = ctx.channel();
		Channel findChannel = wssChannelGroup.find(receiverChannel.id());
		if (findChannel != null) {
			// 用户在线
			receiverChannel.writeAndFlush(new TextWebSocketFrame(cause.getMessage()));
		}
		// 发生异常之后关闭channel
		ctx.channel().close();
		// 随后从ChannelGroup中移除对应的channel
		wssChannelGroup.remove(ctx.channel());
		String channelId = ctx.channel().id().asLongText();
		// 移除多余会话
		removeIdentityId(channelId);
	}

	public Long getIdentityIdByToken(String token){
		if(StrUtil.isBlank(token)){
			GraceException.display(ResponseType.FORBIDDEN);
		}
		LoginUser loginUser = tokenUtil.getLoginUserByToken(token);
		return loginUser.getIdentityId();
	}

	/**
	 * 移除多余会话
	 * @param channelId 信道ID
	 */
	public void removeIdentityId(String channelId){
		Long identityId = UserChannelSession.getIdentityIdByChannelId(channelId);
		Set<Long> deviceIdList = redisUtil.getCacheObject(BrokerConstants.IDENTITY_SUB_DEVICE + identityId);
		redisUtil.deleteObject(BrokerConstants.IDENTITY_SUB_DEVICE + identityId);
		if(deviceIdList != null){
			for(Long deviceId : deviceIdList){
				Set<Long> identityIdList = redisUtil.getCacheObject(BrokerConstants.DEVICE_SUB_IDENTITY + deviceId);
				if(identityIdList != null){
					identityIdList.remove(identityId);
					if(identityIdList.isEmpty()){
						redisUtil.deleteObject(BrokerConstants.DEVICE_SUB_IDENTITY + deviceId);
					}else {
						redisUtil.setCacheObject(BrokerConstants.DEVICE_SUB_IDENTITY + deviceId, identityIdList);
					}
				}
			}
			UserChannelSession.removeChannel(identityId, channelId);
		}
	}

	@PreDestroy
	public void clearRedisKey() {
		List<String> identityKeys = redisUtil.getKeys(BrokerConstants.IDENTITY_SUB_DEVICE + "*");
		for(String identityKey :identityKeys){
			redisUtil.deleteObject(identityKey);
		}
		List<String> deviceKeys = redisUtil.getKeys(BrokerConstants.DEVICE_SUB_IDENTITY + "*");
		for(String deviceKey :deviceKeys){
			redisUtil.deleteObject(deviceKey);
		}
		log.info("Redis identityKey and deviceKey has been deleted");
	}
}
